package com.cnpc.bds.sdk.sink.kafka.handler;

/**
 * 任务链封装.
 */
@SuppressWarnings({"rawtypes", "unchecked"})
public abstract class AbstractPipeline implements Pipeline {


    protected AbstractHandler headHandler = new HeadHandler();
    protected AbstractHandler tailHandler = headHandler;

    @Override
    public AbstractPipeline addLast(Handler e) {
        addLast0((AbstractHandler) e);
        return self();
    }

    @Override
    public AbstractPipeline addFirst(Handler e) {
        addFirst0((AbstractHandler) e);
        return self();
    }

    @Override
    public AbstractHandler head() {
        return headHandler;
    }

    @Override
    public void firePipeline(Object msg) {
        head().doHandle(msg);
    }


    protected AbstractPipeline self() {
        return this;
    }

    private void addLast0(AbstractHandler next) {
        AbstractHandler tail = this.tailHandler;
        tail.next(next);
        next.prev(tail);
        this.tailHandler = next;
    }

    private void addFirst0(AbstractHandler prev) {
        AbstractHandler head = this.headHandler;
        head.prev(prev);
        prev.next(head);
        this.headHandler = prev;
    }

    /**
     * 责任链头节点.
     */
    static class HeadHandler extends AbstractHandler {

        public Object handle(Object msg) {
            //do nothing
            return msg;
        }
    }

}
